//! # 事件驱动系统
//!
//! 提供完整的事件驱动架构支持，包括事件定义、事件处理器、事件路由和事件聚合

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use uuid::Uuid;

use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;

/// 事件类型枚举
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EventType {
    /// 领域事件 - 表示业务状态变化
    Domain,
    /// 集成事件 - 用于系统间通信
    Integration,
    /// 通知事件 - 用于通知和提醒
    Notification,
    /// 系统事件 - 用于系统级操作
    System,
    /// 错误事件 - 用于错误和异常处理
    Error,
    /// 性能事件 - 用于性能监控
    Performance,
    /// 审计事件 - 用于审计和日志
    Audit,
}

/// 事件优先级
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EventPriority {
    /// 低优先级
    Low,
    /// 普通优先级
    Normal,
    /// 高优先级
    High,
    /// 紧急优先级
    Critical,
}

impl Default for EventPriority {
    fn default() -> Self {
        EventPriority::Normal
    }
}

/// 事件上下文信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventContext {
    /// 追踪ID
    pub trace_id: String,
    /// 关联ID（用于关联相关事件）
    pub correlation_id: String,
    /// 用户ID
    pub user_id: Option<String>,
    /// 会话ID
    pub session_id: Option<String>,
    /// 租户ID（多租户场景）
    pub tenant_id: Option<String>,
    /// 源服务
    pub source_service: String,
    /// 目标服务
    pub target_service: Option<String>,
    /// 附加属性
    pub properties: HashMap<String, String>,
}

impl EventContext {
    /// 创建新的事件上下文
    pub fn new(source_service: String) -> Self {
        Self {
            trace_id: Uuid::new_v4().to_string(),
            correlation_id: Uuid::new_v4().to_string(),
            user_id: None,
            session_id: None,
            tenant_id: None,
            source_service,
            target_service: None,
            properties: HashMap::new(),
        }
    }

    /// 设置用户ID
    pub fn with_user_id(mut self, user_id: String) -> Self {
        self.user_id = Some(user_id);
        self
    }

    /// 设置会话ID
    pub fn with_session_id(mut self, session_id: String) -> Self {
        self.session_id = Some(session_id);
        self
    }

    /// 设置租户ID
    pub fn with_tenant_id(mut self, tenant_id: String) -> Self {
        self.tenant_id = Some(tenant_id);
        self
    }

    /// 设置目标服务
    pub fn with_target_service(mut self, target_service: String) -> Self {
        self.target_service = Some(target_service);
        self
    }

    /// 添加属性
    pub fn with_property(mut self, key: String, value: String) -> Self {
        self.properties.insert(key, value);
        self
    }

    /// 继承上下文（保持追踪ID和关联ID）
    pub fn inherit(&self, source_service: String) -> Self {
        Self {
            trace_id: self.trace_id.clone(),
            correlation_id: self.correlation_id.clone(),
            user_id: self.user_id.clone(),
            session_id: self.session_id.clone(),
            tenant_id: self.tenant_id.clone(),
            source_service,
            target_service: None,
            properties: self.properties.clone(),
        }
    }
}

/// 事件定义
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Event {
    /// 事件ID
    pub id: String,
    /// 事件名称
    pub name: String,
    /// 事件类型
    pub event_type: EventType,
    /// 事件优先级
    pub priority: EventPriority,
    /// 事件版本
    pub version: String,
    /// 事件数据
    pub data: serde_json::Value,
    /// 事件上下文
    pub context: EventContext,
    /// 事件时间戳
    pub timestamp: chrono::DateTime<chrono::Utc>,
    /// 聚合ID（DDD中的聚合根ID）
    pub aggregate_id: Option<String>,
    /// 聚合类型
    pub aggregate_type: Option<String>,
    /// 事件序列号
    pub sequence: Option<u64>,
    /// 事件元数据
    pub metadata: HashMap<String, String>,
}

impl Event {
    /// 创建新事件
    pub fn new(
        name: String,
        event_type: EventType,
        data: serde_json::Value,
        context: EventContext,
    ) -> Self {
        Self {
            id: Uuid::new_v4().to_string(),
            name,
            event_type,
            priority: EventPriority::default(),
            version: "1.0".to_string(),
            data,
            context,
            timestamp: chrono::Utc::now(),
            aggregate_id: None,
            aggregate_type: None,
            sequence: None,
            metadata: HashMap::new(),
        }
    }

    /// 设置优先级
    pub fn with_priority(mut self, priority: EventPriority) -> Self {
        self.priority = priority;
        self
    }

    /// 设置版本
    pub fn with_version(mut self, version: String) -> Self {
        self.version = version;
        self
    }

    /// 设置聚合信息
    pub fn with_aggregate(mut self, aggregate_id: String, aggregate_type: String) -> Self {
        self.aggregate_id = Some(aggregate_id);
        self.aggregate_type = Some(aggregate_type);
        self
    }

    /// 设置序列号
    pub fn with_sequence(mut self, sequence: u64) -> Self {
        self.sequence = Some(sequence);
        self
    }

    /// 添加元数据
    pub fn with_metadata(mut self, key: String, value: String) -> Self {
        self.metadata.insert(key, value);
        self
    }

    /// 转换为消息总线消息
    pub fn to_message(&self) -> MessageBusResult<Message> {
        let topic = format!("events.{}.{}", self.event_type.to_topic_segment(), self.name);
        let payload = serde_json::to_value(self)
            .map_err(|e| MessageBusError::serialization_error(e.to_string()))?;
        
        let mut message = Message::new(&topic, payload);
        
        // 添加事件相关的消息头
        message.metadata_mut().set_header("event-id".to_string(), self.id.clone());
        message.metadata_mut().set_header("event-type".to_string(), 
            serde_json::to_string(&self.event_type).unwrap_or_default());
        message.metadata_mut().set_header("event-priority".to_string(), 
            serde_json::to_string(&self.priority).unwrap_or_default());
        message.metadata_mut().set_header("trace-id".to_string(), self.context.trace_id.clone());
        message.metadata_mut().set_header("correlation-id".to_string(), self.context.correlation_id.clone());
        
        if let Some(ref aggregate_id) = self.aggregate_id {
            message.metadata_mut().set_header("aggregate-id".to_string(), aggregate_id.clone());
        }
        
        if let Some(ref aggregate_type) = self.aggregate_type {
            message.metadata_mut().set_header("aggregate-type".to_string(), aggregate_type.clone());
        }

        Ok(message)
    }

    /// 从消息总线消息创建事件
    pub fn from_message(message: &Message) -> MessageBusResult<Event> {
        let event: Event = serde_json::from_value(message.payload().clone())
            .map_err(|e| MessageBusError::deserialization_error(e.to_string()))?;
        Ok(event)
    }
}

impl EventType {
    /// 转换为主题段
    fn to_topic_segment(&self) -> &'static str {
        match self {
            EventType::Domain => "domain",
            EventType::Integration => "integration",
            EventType::Notification => "notification",
            EventType::System => "system",
            EventType::Error => "error",
            EventType::Performance => "performance",
            EventType::Audit => "audit",
        }
    }
}

/// 事件处理器接口
#[async_trait]
pub trait EventHandler: Send + Sync {
    /// 处理事件
    async fn handle(&self, event: &Event) -> MessageBusResult<()>;

    /// 获取处理器名称
    fn name(&self) -> &str;

    /// 获取支持的事件类型
    fn supported_events(&self) -> Vec<String>;

    /// 检查是否可以处理该事件
    fn can_handle(&self, event: &Event) -> bool {
        self.supported_events().contains(&event.name) ||
        self.supported_events().iter().any(|pattern| {
            pattern.contains('*') && self.matches_pattern(pattern, &event.name)
        })
    }

    /// 模式匹配
    fn matches_pattern(&self, pattern: &str, event_name: &str) -> bool {
        if pattern == "*" {
            return true;
        }
        
        if pattern.ends_with("*") {
            let prefix = &pattern[..pattern.len() - 1];
            return event_name.starts_with(prefix);
        }
        
        if pattern.starts_with("*") {
            let suffix = &pattern[1..];
            return event_name.ends_with(suffix);
        }
        
        pattern == event_name
    }
}

/// 事件总线接口
#[async_trait]
pub trait EventBus: Send + Sync {
    /// 发布事件
    async fn publish_event(&self, event: Event) -> MessageBusResult<()>;

    /// 订阅事件
    async fn subscribe_event(&self, pattern: &str, handler: Arc<dyn EventHandler>) -> MessageBusResult<String>;

    /// 取消事件订阅
    async fn unsubscribe_event(&self, subscription_id: &str) -> MessageBusResult<()>;

    /// 批量发布事件
    async fn publish_events(&self, events: Vec<Event>) -> MessageBusResult<()>;

    /// 获取事件统计信息
    async fn get_event_stats(&self) -> MessageBusResult<EventStats>;
}

/// 事件统计信息
#[derive(Debug, Clone, Default)]
pub struct EventStats {
    /// 总发布事件数
    pub total_published: u64,
    /// 总处理事件数
    pub total_processed: u64,
    /// 总失败事件数
    pub total_failed: u64,
    /// 活跃处理器数
    pub active_handlers: usize,
    /// 按类型分组的事件数
    pub events_by_type: HashMap<EventType, u64>,
    /// 按优先级分组的事件数
    pub events_by_priority: HashMap<EventPriority, u64>,
    /// 平均处理时间（毫秒）
    pub average_processing_time_ms: f64,
}

/// 事件路由器
pub struct EventRouter {
    handlers: RwLock<HashMap<String, Arc<dyn EventHandler>>>,
    subscriptions: RwLock<HashMap<String, Vec<String>>>, // pattern -> handler_ids
    stats: Arc<RwLock<EventStats>>,
}

impl EventRouter {
    /// 创建新的事件路由器
    pub fn new() -> Self {
        Self {
            handlers: RwLock::new(HashMap::new()),
            subscriptions: RwLock::new(HashMap::new()),
            stats: Arc::new(RwLock::new(EventStats::default())),
        }
    }

    /// 注册事件处理器
    pub async fn register_handler(&self, handler: Arc<dyn EventHandler>) -> MessageBusResult<String> {
        let handler_id = Uuid::new_v4().to_string();
        let handler_name = handler.name().to_string();
        
        {
            let mut handlers = self.handlers.write().await;
            handlers.insert(handler_id.clone(), handler.clone());
        }

        {
            let mut stats = self.stats.write().await;
            stats.active_handlers += 1;
        }

        tracing::info!("注册事件处理器: {} ({})", handler_name, handler_id);
        Ok(handler_id)
    }

    /// 注销事件处理器
    pub async fn unregister_handler(&self, handler_id: &str) -> MessageBusResult<()> {
        let removed = {
            let mut handlers = self.handlers.write().await;
            handlers.remove(handler_id)
        };

        if removed.is_some() {
            // 清理订阅
            let mut subscriptions = self.subscriptions.write().await;
            for (_, handler_ids) in subscriptions.iter_mut() {
                handler_ids.retain(|id| id != handler_id);
            }
            subscriptions.retain(|_, handler_ids| !handler_ids.is_empty());

            let mut stats = self.stats.write().await;
            stats.active_handlers = stats.active_handlers.saturating_sub(1);

            tracing::info!("注销事件处理器: {}", handler_id);
            Ok(())
        } else {
            Err(MessageBusError::subscribe_error(handler_id, "处理器不存在"))
        }
    }

    /// 订阅事件模式
    pub async fn subscribe(&self, pattern: &str, handler_id: &str) -> MessageBusResult<()> {
        // 检查处理器是否存在
        {
            let handlers = self.handlers.read().await;
            if !handlers.contains_key(handler_id) {
                return Err(MessageBusError::subscribe_error(handler_id, "处理器不存在"));
            }
        }

        {
            let mut subscriptions = self.subscriptions.write().await;
            subscriptions
                .entry(pattern.to_string())
                .or_insert_with(Vec::new)
                .push(handler_id.to_string());
        }

        tracing::debug!("事件订阅: {} -> {}", pattern, handler_id);
        Ok(())
    }

    /// 取消订阅
    pub async fn unsubscribe(&self, pattern: &str, handler_id: &str) -> MessageBusResult<()> {
        {
            let mut subscriptions = self.subscriptions.write().await;
            if let Some(handler_ids) = subscriptions.get_mut(pattern) {
                handler_ids.retain(|id| id != handler_id);
                if handler_ids.is_empty() {
                    subscriptions.remove(pattern);
                }
            }
        }

        tracing::debug!("取消事件订阅: {} -> {}", pattern, handler_id);
        Ok(())
    }

    /// 路由事件到处理器
    pub async fn route_event(&self, event: &Event) -> MessageBusResult<()> {
        let start_time = std::time::Instant::now();
        let matching_handlers = self.find_matching_handlers(event).await;
        
        if matching_handlers.is_empty() {
            tracing::debug!("没有找到匹配的事件处理器: {}", event.name);
            return Ok(());
        }

        let mut success_count = 0;
        let mut error_count = 0;

        // 并行处理
        let tasks: Vec<_> = matching_handlers.into_iter().map(|handler| {
            let event = event.clone();
            tokio::spawn(async move {
                match handler.handle(&event).await {
                    Ok(()) => {
                        tracing::debug!("事件处理成功: {} -> {}", event.name, handler.name());
                        Ok(())
                    }
                    Err(e) => {
                        tracing::error!("事件处理失败: {} -> {}: {}", event.name, handler.name(), e);
                        Err(e)
                    }
                }
            })
        }).collect();

        // 等待所有处理器完成
        for task in tasks {
            match task.await {
                Ok(Ok(())) => success_count += 1,
                Ok(Err(_)) | Err(_) => error_count += 1,
            }
        }

        // 更新统计信息
        let processing_time = start_time.elapsed().as_millis() as f64;
        {
            let mut stats = self.stats.write().await;
            stats.total_processed += success_count;
            stats.total_failed += error_count;
            
            // 更新按类型统计
            *stats.events_by_type.entry(event.event_type.clone()).or_insert(0) += 1;
            
            // 更新按优先级统计
            *stats.events_by_priority.entry(event.priority.clone()).or_insert(0) += 1;
            
            // 更新平均处理时间
            let total_events = stats.total_processed + stats.total_failed;
            if total_events > 0 {
                stats.average_processing_time_ms = 
                    (stats.average_processing_time_ms * (total_events - 1) as f64 + processing_time) / total_events as f64;
            }
        }

        if error_count > 0 {
            Err(MessageBusError::internal_error(format!(
                "事件处理部分失败: 成功 {}, 失败 {}", success_count, error_count
            )))
        } else {
            Ok(())
        }
    }

    /// 查找匹配的处理器
    async fn find_matching_handlers(&self, event: &Event) -> Vec<Arc<dyn EventHandler>> {
        let mut matching_handlers = Vec::new();
        
        let handlers = self.handlers.read().await;
        let subscriptions = self.subscriptions.read().await;

        for (pattern, handler_ids) in subscriptions.iter() {
            if self.event_matches_pattern(event, pattern) {
                for handler_id in handler_ids {
                    if let Some(handler) = handlers.get(handler_id) {
                        if handler.can_handle(event) {
                            matching_handlers.push(handler.clone());
                        }
                    }
                }
            }
        }

        matching_handlers
    }

    /// 检查事件是否匹配模式
    fn event_matches_pattern(&self, event: &Event, pattern: &str) -> bool {
        if pattern == "*" {
            return true;
        }

        // 支持事件名称匹配
        if pattern.contains('*') {
            return self.wildcard_match(pattern, &event.name);
        }

        // 支持事件类型匹配
        if pattern.starts_with("type:") {
            let event_type_pattern = &pattern[5..];
            return event_type_pattern == event.event_type.to_topic_segment();
        }

        // 支持优先级匹配
        if pattern.starts_with("priority:") {
            let priority_pattern = &pattern[9..];
            return priority_pattern == format!("{:?}", event.priority).to_lowercase();
        }

        // 默认匹配事件名称
        pattern == event.name
    }

    /// 通配符匹配
    fn wildcard_match(&self, pattern: &str, text: &str) -> bool {
        if pattern == "*" {
            return true;
        }
        
        if pattern.ends_with("*") {
            let prefix = &pattern[..pattern.len() - 1];
            return text.starts_with(prefix);
        }
        
        if pattern.starts_with("*") {
            let suffix = &pattern[1..];
            return text.ends_with(suffix);
        }
        
        pattern == text
    }

    /// 获取统计信息
    pub async fn get_stats(&self) -> EventStats {
        let stats = self.stats.read().await;
        stats.clone()
    }

    /// 更新发布统计
    pub async fn update_published_stats(&self) {
        let mut stats = self.stats.write().await;
        stats.total_published += 1;
    }
}

impl Default for EventRouter {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[tokio::test]
    async fn test_event_creation() {
        let context = EventContext::new("test-service".to_string())
            .with_user_id("user123".to_string())
            .with_session_id("session456".to_string());

        let event = Event::new(
            "user.created".to_string(),
            EventType::Domain,
            json!({"user_id": "123", "email": "test@example.com"}),
            context,
        )
        .with_priority(EventPriority::High)
        .with_aggregate("user123".to_string(), "User".to_string());

        assert_eq!(event.name, "user.created");
        assert_eq!(event.event_type, EventType::Domain);
        assert_eq!(event.priority, EventPriority::High);
        assert_eq!(event.aggregate_id, Some("user123".to_string()));
    }

    #[tokio::test]
    async fn test_event_message_conversion() {
        let context = EventContext::new("test-service".to_string());
        let event = Event::new(
            "order.created".to_string(),
            EventType::Domain,
            json!({"order_id": "order123"}),
            context,
        );

        // 转换为消息
        let message = event.to_message().unwrap();
        assert_eq!(message.topic(), "events.domain.order.created");

        // 从消息恢复事件
        let recovered_event = Event::from_message(&message).unwrap();
        assert_eq!(recovered_event.name, event.name);
        assert_eq!(recovered_event.event_type, event.event_type);
    }

    #[tokio::test]
    async fn test_event_router() {
        let router = EventRouter::new();

        // 创建测试处理器
        struct TestHandler {
            name: String,
            events: Vec<String>,
        }

        #[async_trait]
        impl EventHandler for TestHandler {
            async fn handle(&self, _event: &Event) -> MessageBusResult<()> {
                Ok(())
            }

            fn name(&self) -> &str {
                &self.name
            }

            fn supported_events(&self) -> Vec<String> {
                self.events.clone()
            }
        }

        let handler = Arc::new(TestHandler {
            name: "test-handler".to_string(),
            events: vec!["user.*".to_string()],
        });

        // 注册处理器
        let handler_id = router.register_handler(handler).await.unwrap();

        // 订阅事件
        router.subscribe("user.*", &handler_id).await.unwrap();

        // 创建事件
        let context = EventContext::new("test-service".to_string());
        let event = Event::new(
            "user.created".to_string(),
            EventType::Domain,
            json!({"user_id": "123"}),
            context,
        );

        // 路由事件
        assert!(router.route_event(&event).await.is_ok());

        // 检查统计信息
        let stats = router.get_stats().await;
        assert_eq!(stats.total_processed, 1);
        assert_eq!(stats.active_handlers, 1);
    }
}